package buka.oversea.aeron5.service;

import io.aeron.ExclusivePublication;
import io.aeron.Image;
import io.aeron.Publication;
import io.aeron.cluster.codecs.CloseReason;
import io.aeron.cluster.service.ClientSession;
import io.aeron.cluster.service.Cluster;
import io.aeron.cluster.service.ClusteredService;
import io.aeron.logbuffer.Header;
import org.agrona.DirectBuffer;
import org.agrona.MutableDirectBuffer;
import org.agrona.concurrent.UnsafeBuffer;
import org.springframework.stereotype.Service;

import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class BalanceClusteredService implements ClusteredService {

    private final ConcurrentHashMap<String, Long> balances = new ConcurrentHashMap<>();
    private Cluster cluster;

    @Override
    public void onStart(Cluster cluster, Image snapshotImage) {
        this.cluster = cluster;
    }

    @Override
    public void onSessionOpen(ClientSession clientSession, long l) {

    }

    @Override
    public void onSessionClose(ClientSession clientSession, long l, CloseReason closeReason) {

    }

    @Override
    public void onSessionMessage(ClientSession session,
                                 long timestamp,
                                 DirectBuffer buffer,
                                 int offset,
                                 int length,
                                 Header header) {
        byte op = buffer.getByte(offset);             // 0 add 1 sub
        long amount = buffer.getLong(offset + 1);
        String account = buffer.getStringAscii(offset + 9, length - 9);

        long newBalance = op == 0
                ? balances.getOrDefault(account, 0L) + amount
                : balances.getOrDefault(account, 0L) - amount;
        balances.put(account, newBalance);

        MutableDirectBuffer resp = new UnsafeBuffer(ByteBuffer.allocate(8));
        resp.putLong(0, newBalance);
        if (session != null) session.offer(resp, 0, 8);
    }

    @Override
    public void onTimerEvent(long l, long l1) {

    }

    @Override
    public void onTakeSnapshot(ExclusivePublication exclusivePublication) {

    }

    public Long balance(String account) {
        return balances.get(account);
    }

    /* 其余生命周期留空即可 */
    @Override public void onRoleChange(Cluster.Role newRole) {}
    @Override public void onTerminate(Cluster cluster) {}
}
